package com.ufo.threads;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class AsyncSaveQueue {
    static AtomicLong atomicLong=new AtomicLong();
    public static void main(String[] args){
        ExecutorService pool = new ThreadPoolExecutor(
                10,
                40,
                300, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(128),
                new ThreadPoolExecutor.AbortPolicy());
        ExecutorService pool2 = new ThreadPoolExecutor(
                4,
                20,
                3600, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(64),
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i=0;i<1000;i++){
            System.out.println(i);
            try{
                CompletableFuture.supplyAsync(()->{
                    try {
                        System.out.println("do some "+atomicLong.incrementAndGet());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return "down";
                },pool).thenAccept(res->{System.out.println("do some down");});
            }catch (Exception e){
                e.printStackTrace();
                CompletableFuture.supplyAsync(()->{
                    try {
                        System.out.println("补录");
                        Thread.sleep(1000);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                    return "down";
                },pool2).thenAccept(res->{
                    System.out.println("死信完毕");
                });
            }
        }
        System.out.println("结束");
        ThreadPoolExecutor tpe = ((ThreadPoolExecutor) pool);
        ThreadPoolExecutor tpe2 = ((ThreadPoolExecutor) pool2);
        System.out.println("剩余任务："+tpe.getQueue().size());
        while(true){
            if (tpe2.getQueue().size()<1){
                System.out.println("手动");
                System.out.println(tpe2.getQueue().size() +":"+tpe.getQueue().size());
                tpe2.shutdown();
                tpe.shutdown();
                break;
            }
        }
    }
}
